package com.yanggu.flink.datastream_api.process_function.timerservice

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.Event
import com.yanggu.flink.datastream_api.source.ClickSource
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object TopNDemo02 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env
      .addSource(new ClickSource())
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(element: Event, recordTimestamp: Long) = element.timestamp
        })
      )
      .keyBy(data => data.url)
      .window(SlidingEventTimeWindows.of(Time.seconds(10L), Time.seconds(1L)))
      .aggregate(new AggregateFunction[Event, Long, Long] {
        override def createAccumulator() = 0L

        override def add(value: Event, accumulator: Long) = accumulator + 1L

        override def getResult(accumulator: Long) = accumulator

        override def merge(a: Long, b: Long) = a + b
      }, new ProcessWindowFunction[Long, (String, Long, Long), String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[Long], out: Collector[(String, Long, Long)]): Unit = {
          val count = elements.head
          out.collect((key, count, context.window.getEnd))
        }
      })
      .keyBy(tuple => tuple._3)
      .process(new KeyedProcessFunction[Long, (String, Long, Long), String] {

        private var listState: ListState[(String, Long, Long)] = _

        override def open(parameters: Configuration): Unit = {
          listState = getRuntimeContext.getListState(new ListStateDescriptor[(String, Long, Long)]("listState", classOf[(String, Long, Long)]))
        }

        override def processElement(value: (String, Long, Long), ctx: KeyedProcessFunction[Long, (String, Long, Long), String]#Context, out: Collector[String]): Unit = {
          listState.add(value)
          ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey + 1)
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, (String, Long, Long), String]#OnTimerContext, out: Collector[String]): Unit = {
          //有界优先队列 => TopN

          val list = new ArrayBuffer[(String, Long, Long)]
          listState.get().forEach(tuple => list.append(tuple))

          val tuples = list
            //按照访问次数升序排序
            .sortBy(tuple =>  - tuple._2)
            //从右开始取5个
            .takeRight(5)

          val sb = new mutable.StringBuilder()
          sb.append("访问次数top10\r\n")

          for (index <- tuples.indices) {
            val data = tuples(index)
            sb.append(s"${index + 1}. url: ${data._1}, 访问次数: ${data._2}, " +
              s"开始时间: ${DateUtil.formatDateTime(new Date(data._3 - 10000))}, " +
              s"结束时间: ${DateUtil.formatDateTime(new Date(data._3))} \r\n")
          }

          out.collect(sb.toString())

          listState.clear()
        }

      })
      .print()

    env.execute()

  }

}
